package com.zt.bigdata.spark.scala.asyn

import java.util.concurrent.{ExecutorService, Executors}

import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Failure, Random}
//import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.Success


object ScalaAsyn1 extends App {


  implicit lazy val workStealingPoolExecutionContext: ExecutionContext = {
    val workStealingPool: ExecutorService = Executors.newWorkStealingPool
    ExecutionContext.fromExecutor(workStealingPool)
  }

  var start = System.currentTimeMillis()
  val value1 = {
    Thread.sleep(1000)
    1
  }
  var end = System.currentTimeMillis()
  println(s"同步等待了${end - start} 毫秒,返回值:$value1")

  start = System.currentTimeMillis()
  val value2 = Future {
    Thread.sleep(2000)
    1
  }

  end = System.currentTimeMillis()

  println(s"异步等待了${end - start} 毫秒,返回值:${value2.value}")


  import scala.concurrent.Await
  import scala.concurrent.duration._

  val resultValue = Await.result(value2, 2 seconds)
  println(s"阻塞方法,返回值:${resultValue}")


  val value3 = Future {
    val p = Promise[String]
    val random = new Random()
    if (random.nextInt % 2 == 0) p.success("success") else p.failure(null)
    p.future
  }

  value3.onComplete {
    case Success(intValue) => println("success:   " + intValue)
    case Failure(error) => println("An error has occured: " + error.getMessage)
  }


  def future(f: String): Future[String] = {
    val future = Promise[String]
    //todo .....
    future.success("error")
    future.future
  }

  val f = future("hello")

  f.onSuccess {
    case "hello" => print("world")
    case _ => print("error")
  }


  def recordExists(id: Long): Boolean = {
    println(s"recordExists($id)...")
    Thread.sleep(1)
    id > 0
  }

  def getRecord(id: Long): (Long, String) = {
    println(s"getRecord:($id)...")
    Thread.sleep(1)
    (id, s"record: $id")
  }

  def asyncGetRecord(id: Long): Future[(Long, String)] = Future {
    val exists: Future[Boolean] = Future {
      val b = recordExists(id);
      println(b)
      b
    }
    val value = Await.result(exists, 2 seconds)
    if (value) {
      getRecord(id)
    } else {
      (id, "record not found")
    }
  }


  asyncGetRecord(1).onComplete {
    case Success(kv) => println(kv)
  }

  /*
    val pool: ExecutorService = Executors.newCachedThreadPool()
    val future = new FutureTask[String](new Callable[String]() {
      def call(): String = {
        Thread.sleep(2000)
        "success"
      }
    })
     pool.execute(future)
    val blockingResult = Await.result(future, 3 second)

  */

}
